package com.ksc.wordcount.shuffle.nettyimpl.server;

import com.ksc.wordcount.conf.AppConfig;
import com.ksc.wordcount.shuffle.ShuffleBlockId;
import com.ksc.wordcount.shuffle.nettyimpl.FileComplate;
import io.netty.channel.*;
import java.io.EOFException;
import java.io.File;
import java.io.FileInputStream;
import java.io.ObjectInputStream;

public class ShuffleServiceHandler extends ChannelInboundHandlerAdapter {

    /**
     * 当收到客户端请求时调用该方法
     *
     * @param ctx ChannelHandlerContext对象，提供操作Channel的方法
     * @param msg 接收到的消息对象
     */
    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
        System.out.println("ShuffleServiceHandler received:" + msg);

        // 判断接收到的消息是否为ShuffleBlockId类型
        if (msg instanceof ShuffleBlockId) {
            ShuffleBlockId shuffleBlockId = (ShuffleBlockId) msg;
            System.out.println("ShuffleServiceHandler received name:" + ((ShuffleBlockId) msg).name());

//            System.out.println("ShuffleServiceHandler received:" + ((ShuffleBlockId) msg).name());
            File file = new File(shuffleBlockId.getShufflePath());

            // 检查对应的文件是否存在
            if (file.exists()) {
                ObjectInputStream objectInputStream = new ObjectInputStream(new FileInputStream(file));
                Object obj = null;

                // 持续从文件中读取对象，直到文件结尾
                do {
                    try {
                        obj = objectInputStream.readObject();
                    } catch (EOFException e) {
                        break;
                    }

                    // 将读取到的对象写入到Channel，并刷新到对应的客户端
                    ctx.writeAndFlush(obj);
                } while (obj != null);

                System.out.println("ShuffleServiceHandler send FileComplate");

                // 写入FileComplate对象到Channel，并刷新到对应的客户端
                ctx.writeAndFlush(new FileComplate());
            } else {
                // 如果文件不存在，则向客户端写入错误信息
                ctx.writeAndFlush("shuffle File not found: " + file.getAbsolutePath());
            }
        } else {
            // 如果接收到的不是ShuffleBlockId类型的消息，则调用父类的channelRead方法继续处理
            super.channelRead(ctx, msg);
        }
    }

    /**
     * 异常捕获方法，当发生异常时调用该方法
     *
     * @param ctx    ChannelHandlerContext对象，提供操作Channel的方法
     * @param cause  异常对象
     */
    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
        cause.printStackTrace();
        ctx.close();
    }
}